package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.utils.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 从Kafka读取业务数据，封装为FLink SQL 表
        String topicName = "topic_db";
        String groupID = "dwd_trade_order_pre_process";
        tableEnv.executeSql("create table topic_db(\n" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<string,String>,\n" +
                "`old` map<String,String>,\n" +
                "`ts` String\n" +
                ")" + KafkaUtil.getKafkaDDL(topicName, groupID));

        //TODO 3 读取订单明细表数据
        Table orderDetail = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['origin_amount'] origin_amount,\n" +
                "`data`['coupon_reduce'] coupon_reduce,\n" +
                "`data`['final_amount'] final_amount,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`ts` od_ts\n" +
                "from\n" +
                "`topic_db`\n" +
                "where\n" +
                "`table` = 'order_detail'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("order_detail",orderDetail);

        //TODO 4 读取订单表数据
        Table orderInfo = tableEnv.sqlQuery("select\n" +
                "`old`,\n" +
                "`type`,\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['session_id'] session_id,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`ts` oi_ts\n" +
                "from\n" +
                "`topic_db`\n" +
                "where\n" +
                "`table` = 'order_info'\n" +
                "and (`type` = 'insert' or 'type' = 'update')");
        tableEnv.createTemporaryView("order_info",orderInfo);

        //TODO 5 读取支付表数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                "`old`,\n" +
                "`type`,\n" +
                "`data`['id'] id,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['alipay_trade_no'] alipay_trade_no,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['payment_type'] payment_type,\n" +
                "`data`['payment_status'] payment_status,\n" +
                "`data`['callback_content'] callback_content,\n" +
                "`data`['callback_time'] callback_time,\n" +
                "`ts` pi_ts\n" +
                "from\n" +
                "`topic_db`\n" +
                "where\n" +
                "`table` = 'payment_info'\n" +
                "and `type` = 'insert'");
        tableEnv.createTemporaryView("payment_info",paymentInfo);

        //TODO 6 关联三张表获得订单明细表
        Table resultTable = tableEnv.sqlQuery("select\n" +
                "oi.id id,\n" +
                "course_id,\n" +
                "course_name,\n" +
                "od.user_id user_id,\n" +
                "od.origin_amount origin_amount,\n" +
                "od.coupon_reduce coupon_reduce,\n" +
                "od.final_amount final_amount,\n" +
                "oi.session_id session_id,\n" +
                "od.create_time create_time,\n" +
                "od.update_time update_time,\n" +
                "oi.order_status order_status,\n" +
                "oi.out_trade_no out_trade_no,\n" +
                "oi.trade_body trade_body,\n" +
                "oi.province_id province_id,\n" +
                "oi.expire_time expire_time,\n" +
                "pi.alipay_trade_no alipay_trade_no,\n" +
                "pi.total_amount total_amount,\n" +
                "pi.payment_type payment_type,\n" +
                "pi.payment_status payment_status,\n" +
                "pi.callback_content callback_content,\n" +
                "pi.callback_time callback_time,\n" +
                "oi.`type` type,\n" +
                "oi.`old` `old`,\n" +
                "od.od_ts od_ts,\n" +
                "oi.oi_ts oi_ts\n" +
                "from\n" +
                "order_detail od\n" +
                "join order_info oi\n" +
                "on\n" +
                "od.order_id = oi.id\n" +
                "join payment_info pi\n" +
                "on od.order_id = pi.order_id");
        tableEnv.createTemporaryView("result_table",resultTable);
        resultTable.execute().print();

        //TODO 7 建立 Upsert-Kafka dwd_trade_order_pre_process 表
        tableEnv.executeSql("create table dwd_trade_order_pre_process(\n" +
                "id String,\n" +
                "course_id String,\n" +
                "course_name String,\n" +
                "user_id String,\n" +
                "origin_amount String,\n" +
                "coupon_reduce String,\n" +
                "final_amount String,\n" +
                "session_id String,\n" +
                "create_time String,\n" +
                "update_time String,\n" +
                "order_status String,\n" +
                "out_trade_no String,\n" +
                "trade_body String,\n" +
                "province_id String,\n" +
                "expire_time String,\n" +
                "alipay_trade_no String,\n" +
                "total_amount String,\n" +
                "payment_type String,\n" +
                "payment_status String,\n" +
                "callback_content String,\n" +
                "callback_time String,\n" +
                "`type` String,\n" +
                "`old` map<String,String>,\n" +
                "od_ts String,\n" +
                "oi_ts String,\n" +
                "primary key(id) not enforced\n" +
                ")" + KafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

        //TODO 8 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("insert into dwd_trade_order_pre_process select * from result_table");
//        env.execute("dwd_trade_order_pre_process");
    }
}
